#include <iterator>
#include "MarkupSTL.h"
#include "global.h"
#include "Game.h"
#include "RedisClient.h"
#include "Logic.h"
#include "json.h"
#include "util.h"
#include "globalExtern.h"
#include "Host.h"
#include "CClient.h"
#include "NodeServer.h"
#include "Host.h"
#include <atomic>
#include <thread>
#include "log.h"
#include "json.h"
#include "util.h"
#include "streamdata.h"
#include <functional>
#include "singleton.h"
#include "context.h"
#include "def.h"
#include "minutelisten.h"
#include "2000_topic.pb.h"
#include "LogHandler.h"
#include "ActionTopic.h"


const char* CGameServer::s_szexchangetype = "direct";
const char* CGameServer::s_szexchange = "exceptionflow";


CGameServer::CGameServer(): m_mqconn(NULL), m_mqchannel(1), m_mqPort(0)
{
}

CGameServer::~CGameServer()
{
}

bool CGameServer::InitGame()
{	
	char szconfig[] = "./config.xml";
	if (!ReadCommonConf(szconfig))
		return false;

	if (!initmq())
	{
		log_error("init mq fail");
		return false;
	}

	Ctx.host = m_pHost;

	if (!Ctx.msg_router.Init(TGlobal::_param1))
	{
		log_error("Ctx.msg_router Init failed, port:%d", TGlobal::_param1);
		return false;
	}

	Ctx.timer_event_activator = CActiveEvent::Attach(CCReactor::Instance());
	if (nullptr == Ctx.timer_event_activator)
	{
		log_error("create CActiveEvent fail");
		return -1;
	}
	
	ActionTopic::Init();
	
	// 添加定时任务, 主线程
	Ctx.timer_event_activator->SetSignalHandle([this](std::int64_t event){
		TimerEvent u(event);
		this->StartTimerMillisecond(u.stu.timer_id, u.stu.ms, false);
	});
	// 执行 分钟任务
    Ctx.msg_router.BindTimer(TIMER_MINITE, [](int timer_id)->int{
		MinuteListen::Instance()->OnTick();
        TimerEvent u(timer_id, 60 * 1000);
        Ctx.timer_event_activator->signal(u.event);
		return 0;
    });

	StartTimerMillisecond(TimerID::TIMER_MINITE, 60*1000, false);
	
	return true;
}

int CGameServer::OnBYTimeOut(int Timerid)
{
    Ctx.msg_router.ReceiveTimerData(Timerid);
	return 0;
}

int CGameServer::ProcessPacket(SocketHandler* pHandler, NETInputPacket* inputPkg)
{
	log_debug("cmd=0x%x,uuid:%d", inputPkg->GetCmdType(), inputPkg->GetSeq());

	int cmd = inputPkg->GetCmdType();

	if (0 == LogHandler::Instance()->ProcessPacket(m_pHost, pHandler, inputPkg))
	{
		return 0;
	}
	
	try
	{
		MsgData msg_data;
		msg_data.svtype = pHandler->getType();
		msg_data.svid = pHandler->getId();
		msg_data.from_client = (inputPkg->GetSrcType() == SERVER_TYPE_GAME_PLAYER);
		msg_data.cmd = cmd;
		msg_data.uuid = inputPkg->GetSeq();
		inputPkg->getPbData(msg_data.data);

		Ctx.msg_router.ReceiveMsgData(std::move(msg_data));
	}
	catch(const std::exception& e)
	{
		log_error("cmd=0x%x,error:%s", inputPkg->GetCmdType(), e.what());
		return 0;
	}
	catch(...)
	{
		log_error("cmd=0x%x, unknown exception", inputPkg->GetCmdType());
		return 0;
	}
	
	return 0;
}

int CGameServer::ProcEventBackConn(int ev_id, SpiderEvent* pEvent)
{
	if(pEvent && pEvent->evParam.ptr)
	{
		CClient* client = (CClient*)pEvent->evParam.ptr;
		unsigned short stype =  client->getType();
		unsigned int sid = client->getId();

		if (stype == SERVER_TYPE_AGENCY_EVENT)
		{
			topic_proto::SubscirbReq req;
			req.set_subscrib_type(topic_proto::SubscirbType::INSTANCE);
			req.add_topic_event(topic_proto::TopicEvent::TOPIC_EVENT_AGENCY_EVENT_UPDATE);
			m_pHost->MtSendtoOneSrv(topic_proto::TOPIC_CMD_SUBSCRIB_REQ, &req, stype, sid);
		}
	}
	
	return 0;
}

bool CGameServer::ReadCommonConf(const char* strxml)
{
	CMarkupSTL  markup;
	if (!markup.Load(strxml))
	{
		log_error("Load xml %s failed", strxml);
		return false;
	}

	if (!markup.FindElem("application"))
	{
		log_error("Not Found Elem [application] in xml[%s]", strxml);
		return false;
	}

	if (!markup.IntoElem())
	{
		log_error("IntoElem [application] failed.");
		return false;
	}

	if (!markup.FindElem("mq"))
	{
		log_error("Not Found Elem [mq] in xml[%s]", strxml);
		return false;
	}
	else
	{
		m_mqIp = markup.GetAttrib("ip");
		m_mqPort = atoi(markup.GetAttrib("port").c_str());
		m_mqVhost = markup.GetAttrib("vhost");
		m_mqUsername = markup.GetAttrib("username");
		m_mqPwd = markup.GetAttrib("password");
	}

	if (m_mqIp.empty() || m_mqVhost.empty() || m_mqUsername.empty() || m_mqPwd.empty() || 0 == m_mqPort)
		return false;

	log_debug("ReadCommonConf succ!!");
	return true;
}


bool CGameServer::initmq(void)
{
	m_mqconn = amqp_new_connection();
	if (!m_mqconn)
	{
		log_error("create mq conn err");
		return false;
	}
	amqp_socket_t* socket = amqp_tcp_socket_new(m_mqconn);
	if (!socket)
	{
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
		log_error("create mq TCP socket err");
		return false;
	}
	int status = amqp_socket_open(socket, m_mqIp.c_str(), m_mqPort);
	if (status)
	{
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
		log_error("open mq socket err");
		return false;
	}
	amqp_rpc_reply_t x = amqp_login(m_mqconn, m_mqVhost.c_str(), 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, m_mqUsername.c_str(), m_mqPwd.c_str());
	if (AMQP_RESPONSE_NORMAL != x.reply_type)
	{
		amqp_connection_close(m_mqconn, AMQP_REPLY_SUCCESS);
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
		log_error("mq login fail,host:%s,user:%s,pwd:%s", m_mqVhost.c_str(), m_mqUsername.c_str(), m_mqPwd.c_str());
		return false;
	}
	amqp_channel_open_ok_t* t = amqp_channel_open(m_mqconn, m_mqchannel);
	if (!t)
	{
		amqp_connection_close(m_mqconn, AMQP_REPLY_SUCCESS);
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
		log_error("open mq channel fail");
		return false;
	}

	amqp_rpc_reply_t x1 = amqp_get_rpc_reply(m_mqconn);
	if (AMQP_RESPONSE_NORMAL != x1.reply_type)
	{
		releasemq();
		log_error("mq open channel amqp_get_rpc_reply fail");
		return false;
	}

	amqp_exchange_declare(m_mqconn, m_mqchannel, amqp_cstring_bytes(CGameServer::s_szexchange), amqp_cstring_bytes(s_szexchangetype), false, true, false, false, amqp_empty_table);

	amqp_rpc_reply_t x2 = amqp_get_rpc_reply(m_mqconn);
	if (AMQP_RESPONSE_NORMAL != x2.reply_type)
	{
		releasemq();
		log_error("mq open amqp_exchange_declare fail");
		return false;
	}

	return true;
}

void CGameServer::releasemq(void)
{
	if (m_mqconn)
	{
		amqp_channel_close(m_mqconn, m_mqchannel, AMQP_REPLY_SUCCESS);
		amqp_connection_close(m_mqconn, AMQP_REPLY_SUCCESS);
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
	}

	return;
}